package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.Order_cargo;
import com.bw.bean.Order_info;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;
import java.util.TimeZone;

public class test6 {
    public static void main(String[] args) throws Exception {
        //流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk6", "test672"));

        // 過濾兩張表
//        SingleOutputStreamOperator<String> orderInfo = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"));
//        SingleOutputStreamOperator<String> orderCargo = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_cargo"));


        SingleOutputStreamOperator<String> orderInfo = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    String table = jsonObject.getString("table");
                    if ("order_info".equals(table)) {
                        return true;
                    }
                } catch (Exception e) {
                }
                return false;
            }
        });


        SingleOutputStreamOperator<String> orderCargo = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    String table = jsonObject.getString("table");
                    if ("order_cargo".equals(table)) {
                        return true;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return false;
            }
        });
        orderInfo.print();
        orderCargo.print();

        //1.将order_cargo的数据与order_info数据流通过订单id进行关联。（由于存在数据乱序或者延时的情况，导致两个流的数据进度不一致，
        // 请酌情使用合适的join方式，二者数据前后相差最大5秒，同时order_info状态的ttl设置为5s）（10分）
        //orderInfo.print();
        //orderCargo.print();
        SingleOutputStreamOperator<Order_info> orderInfoDS = orderInfo.map(new MapFunction<String, Order_info>() {
            @Override
            public Order_info map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject data = jsonObject.getJSONObject("data");
                return JSON.parseObject(data.toJSONString(), Order_info.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<Order_info>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                //create_time=2022-07-12T02:54:36Z
                .withTimestampAssigner((event, timestamp) -> {

                    try {
                        SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                        sdf2.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                        long time = sdf2.parse(event.getCreate_time()).getTime();
                        return time;
                    } catch (ParseException e) {
                        e.printStackTrace();
                    }

                    return timestamp;
                }));
        //orderInfoDS.print();

        SingleOutputStreamOperator<Order_cargo> orderCagoDS = orderCargo.map(new MapFunction<String, Order_cargo>() {
            @Override
            public Order_cargo map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject data = jsonObject.getJSONObject("data");
                return JSON.parseObject(data.toJSONString(), Order_cargo.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<Order_cargo>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                //create_time=2022-07-12T02:54:36Z
                .withTimestampAssigner((event, timestamp) -> {
                    try {

                        //处理时区问题
                        SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                        sdf1.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                        long time = sdf1.parse(event.getCreate_time()).getTime();
                        return time;
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));

        // 雙流Join
        SingleOutputStreamOperator<Tuple2<Order_info, Order_cargo>> unionDS = orderInfoDS.keyBy(x -> x.getId()).intervalJoin(orderCagoDS.keyBy(x -> x.getOrder_id())).between(Time.seconds(-5), Time.seconds(0)).process(new ProcessJoinFunction<Order_info, Order_cargo, Tuple2<Order_info, Order_cargo>>() {
            @Override
            public void processElement(Order_info left, Order_cargo right, ProcessJoinFunction<Order_info, Order_cargo, Tuple2<Order_info, Order_cargo>>.Context ctx, Collector<Tuple2<Order_info, Order_cargo>> out) throws Exception {
                out.collect(new Tuple2<>(left, right));
            }
        });
        unionDS.print(">>");
//        (Order_info(id=10061, order_no=2024-1100355484-00000140, status=60020, collect_type=69002,
//         user_id=142, receiver_complex_id=86, receiver_province_id=310000, receiver_city_id=310100,
//         receiver_district_id=310114, receiver_address=上海市市辖区嘉定区湖夏城3号楼123, receiver_name=宇文凡佳,
//         receiver_phone=13422919718, receive_location=32.123123,71.12341, sender_complex_id=80,
//         sender_province_id=520000, sender_city_id=520100, sender_district_id=520102,
//         sender_address=贵州省贵阳市南明区森凯苑3号楼435, sender_name=乐瑞, sender_phone=13267752128,
//         send_location=null, payment_type=62002, cargo_num=1, amount=13.00, estimate_arrive_time=1658181276000,
//         distance=2090.00, create_time=2022-07-12T02:54:36Z, update_time=2022-07-12T02:56:37Z, is_deleted=0),
//         Order_cargo(id=10059, order_id=10061, cargo_type=74002, volume_length=null, volume_width=null,
//         volume_height=null, weight=4.00,
//         remark=null, create_time=2022-07-12T02:54:36Z, update_time=null, is_deleted=0))


        env.execute();
    }
}
